//package com.bsj.sipgateway.mq.consumer;
//
//import com.alibaba.fastjson.JSON;
//import com.bsj.power.common.config.rocketmq.producer.RocketMqConfig;
//import com.bsj.sipgateway.cache.DeviceCache;
//import com.bsj.sipgateway.cache.VehicleCache;
//import com.bsj.sipgateway.command.RocketMqCommand;
//import com.bsj.sipgateway.common.constant.Constants;
//import com.bsj.sipgateway.mq.producer.CommandUpProducer;
//import com.bsj.power.common.def.entity.gb28181.Device;
//import com.bsj.sipgateway.sip.session.VideoStreamSessionManager;
//import com.bsj.sipgateway.sip.transmit.cmd.impl.SIPCommander;
//import com.bsj.sipgateway.util.CommandDownType;
//import com.bsj.util.PropertiesUtil;
//import lombok.extern.slf4j.Slf4j;
//import org.apache.commons.lang3.StringUtils;
//import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
//import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
//import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
//import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
//import org.apache.rocketmq.client.exception.MQClientException;
//import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
//import org.apache.rocketmq.common.message.MessageExt;
//import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
//import org.springframework.beans.factory.annotation.Autowired;
//import org.springframework.stereotype.Component;
//
//import java.util.List;
//
///**
// * @Author WPH
// * @Description 命令下发
// * @Date 2022/8/8 19:35
// */
//@Slf4j
//@Component
//public class CommandDownConsumer {
//    private DefaultMQPushConsumer consumer;
//
//    @Autowired
//    private CommandUpProducer commandUpProducer;
//
//    @Autowired
//    private SIPCommander sipCommander;
//
//    @Autowired
//    private VideoStreamSessionManager videoStreamSessionManager;
//
//    @Autowired
//    private RocketMqConfig mqConfig;
//
//    public void start() throws MQClientException {
//        subscription();
//    }
//
//    public void stop() {
//        if (consumer != null) {
//            consumer.shutdown();
//        }
//    }
//
//    public void subscription() throws MQClientException {
//        //mq地址
//        String rocketmqAddr = PropertiesUtil.getConfig("rocketMQAddr");
//        //消费组
//        String groupName = PropertiesUtil.getConfig("rocketmqCommandDownGroup");
//        String topic = PropertiesUtil.getConfig("rocketmqCommandDownTopic");
//        String tag = "*";
//
//        consumer = new DefaultMQPushConsumer(groupName);
//        consumer.setNamesrvAddr(rocketmqAddr);
//
//        // 从最后开始消费
//        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
//
//        // 广播消费方式
//        consumer.setMessageModel(MessageModel.BROADCASTING);
//        consumer.subscribe(topic, tag);
//        consumer.registerMessageListener(new MessageListenerConcurrently() {
//            @Override
//            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
//                                                            ConsumeConcurrentlyContext context) {
//                for (MessageExt message : msgs) {
//                    if (null == message) {
//                        continue;
//                    }
//                    String msg = new String(message.getBody());
//
//                    RocketMqCommand command = JSON.parseObject(msg, RocketMqCommand.class);
//                    if (null == command) {
//                        log.error("RocketMqCommand is null");
//                        continue;
//                    }
//                    try {
//                        //在vehicleCache里面，说明是GB28181设备，才进行处理
//                        if (VehicleCache.contains(command.getDevice())) {
//                            log.info("mq消费到命令下发指令，{}", msg);
//                            deal(command);
//                        }
//
//                    } catch (Exception e) {
//                        log.error("mq处理命令下发异常", e);
//                        //稍后再试
//                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
//                    }
//                }
//                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
//            }
//        });
//        consumer.start();
//        log.info("开启mq消费命令下发成功");
//    }
//
//    public void deal(RocketMqCommand command) {
//        Device device = DeviceCache.getDevice(command.getDevice());
//        if (device == null) {
//            //设备不在线
//            command.setStatus(0);
//            command.setRetMsg("设备不在线");
//        } else {
//            switch (command.getCmdType()) {
//                case CommandDownType.REAL_MEDIA_OPEN_CMD:
//                case CommandDownType.REAL_AUDIO_OPEN_CMD:
//                case CommandDownType.REAL_VIDEO_OPEN_CMD:
//                    //打开音视频或视频
//                    sipCommander.playStreamCmd(command, command.getDevice());
//                    break;
//                case CommandDownType.DEV_VOD_MEDIA_QUERY:
//                    //设备回放查询指令
//                    String cmdMessage = command.getCmdMessage();
//                    if (StringUtils.isNotBlank(cmdMessage) && cmdMessage.length() >= 26) {
//                        String startTimeBcd = cmdMessage.substring(2, 14);
//                        String endTimeBcd = cmdMessage.substring(14, 26);
//                        log.info("设备{}回放查询，开始时间：{}，结束时间{}", command.getDevice(), startTimeBcd, endTimeBcd);
//
//                        sipCommander.recordInfoQuery(command, startTimeBcd, endTimeBcd);
//                    } else {
//                        command.setStatus(0);
//                        command.setRetMsg("设备回放查询指令参数错误");
//                        log.error("设备回放查询指令参数错误，{}", JSON.toJSONString(command));
//                    }
//                    break;
//                case CommandDownType.DEV_VOD_MEDIA_TRANS:
//                    //设备回放传输指令
//                    sipCommander.playbackStreamCmd(command, command.getDevice());
//                    break;
//                case CommandDownType.REAL_INTERCOM_OPEN_CMD:
//                    //打开对讲
//                    if (videoStreamSessionManager.contain(command.getDevice(), Constants.BROADCAST_PREFIX, command.getSpareMessage())) {
//                        //已经打开对讲，mq响应失败
//                        command.setStatus(0);
//                        command.setRetMsg("设备正在对讲中");
//                    } else {
//                        //先下发实时视频点播，等ACK后再下发语音广播
//                        sipCommander.playStreamCmd(command, command.getDevice());
//                    }
//                    break;
//                case CommandDownType.REAL_MEDIA_CTRL_CLOSEALL:
//                case CommandDownType.REAL_MEDIA_CTRL_CLOSEINTERCOM:
//                    //关闭实时视频
//                    sipCommander.streamByeCmd(device.getDeviceId(), Constants.PLAYER_PREFIX, command);
//                    break;
//                case CommandDownType.DEV_VOD_MEDIA_CLOSE:
//                    //关闭历史视频
//                    sipCommander.streamByeCmd(device.getDeviceId(), Constants.PLAY_BLACK_PREFIX, command);
//                    break;
//                case CommandDownType.PTZ_ROTATE:
//                case CommandDownType.PTZ_ADJUSTING_FOCAL_CTRL:
//                case CommandDownType.PTZ_ADJUSTING_APERTURE_CTRL:
//                case CommandDownType.PTZ_WIPER_CTRL:
//                case CommandDownType.PTZ_ZOOM_CTRL:
//                    sipCommander.ptz(command, command.getDevice());
//                    break;
//            }
//        }
//        //响应mq
//        commandUpProducer.send(command, Constants.SQL_INSERT);
//    }
//
//}
